package common

import org.apache.hadoop.hbase.{HBaseConfiguration, HColumnDescriptor, TableName}
import org.apache.hadoop.hbase.client.{ConnectionFactory, Result, Table}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.SparkContext

object Utils {
  val hbase_zq = "10.120.65.181,10.120.67.164,10.120.64.243,10.120.64.144"
  val hbase_zqPort = "2181"
  val hBase_zParent = "/hbase"
  val configuration = HBaseConfiguration.create()
  this.configuration.set("hbase.zookeeper.quorum", this.hbase_zq)
  this.configuration.set("hbase.zookeeper.property.clientPort", this.hbase_zqPort)
  this.configuration.set("zookeeper.znode.parent", this.hBase_zParent)
  val connection = ConnectionFactory.createConnection(this.configuration)
  val admin = connection.getAdmin()

  def curTable(tablename: String): Table = {
    val tName = TableName.valueOf(tablename)
    this.connection.getTable(tName)
  }
  def get_job(sc:SparkContext,tablename:String): Job = {
    sc.hadoopConfiguration.set("hbase.zookeeper.quorum", hbase_zq)
    sc.hadoopConfiguration.set("hbase.zookeeper.property.client", hbase_zqPort)
    sc.hadoopConfiguration.set(TableOutputFormat.OUTPUT_TABLE, tablename)
    val job: Job = Job.getInstance(sc.hadoopConfiguration)
    job.setOutputKeyClass(classOf[ImmutableBytesWritable])
    job.setOutputValueClass(classOf[Result])
    job.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])
    job
  }

  def addFamilys(tableName: String, columnFamilys: String): Unit = {
    val tablename = TableName.valueOf(tableName)
    this.admin.disableTable(tablename)
    val hDescriptor = this.admin.getTableDescriptor(tablename)
    val hColumnDescriptor = new HColumnDescriptor(columnFamilys)
    val descriptor = hDescriptor.getFamily(Bytes.toBytes(columnFamilys))
    if (descriptor == null) {
      hDescriptor.addFamily(hColumnDescriptor)
      this.admin.modifyTable(tablename, hDescriptor)
    }
    this.admin.enableTableAsync(tablename)
  }
}


